package day09

import org.apache.spark.{HashPartitioner, SparkConf}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object AccSparkStreamingWC {
  /**
    * 因为streaming是批次执行的，因此当我们要用到历史的批次结果时：可以用批次累加函数updateStateByKey( , , )原语实现。（另一种方式是将批次的结果存储到数据库中调用）
    * updateStateByKey():
    *   第一个参数是func,第二个参数是自定义一个分区器(分区器参数可以是默认分区，可以自己设置分区数)，第三个true
    * 该函数使用注意：该函数只有获取历史批次结果引用到当前批次中的功能，是没有存储历史批次结果的功能的；
    *     所以，实现批次累加函数必须要提前先进行checkpoint，streaming中checkpoint具有存储历史结果的功能
    * 第一个参数func函数说明：func函数的输入it参数说明：(it: Iterator[(String, Seq[Int], Option[Int])])
    *   key ：
    *   Iterator(v) : 当前批次相同key对应的V  如：(1,1,1)
    *   POME(v): 历史结果中相同key对应的value     如：Some(2)
    *
    * 注意，能用sum就别用reduce()，因为当批次输入为空的时候用reduce聚合会报错
    *
    */
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "root")
    val conf: SparkConf = new SparkConf().setAppName("accsparkstreamingwc").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(5))

    /*
    * SparkStreaming的checkpoint会报用户权限错误，因为我在IDEA开发程序连接hdfs因为本机名和服务器文件所属用户名不同
    * 导致没有读取权限，报错为：AccessControlException: Permission denied: user=fuhao, access=WRITE, inode="/cp-2020-7-15":root:supergroup:drwxr-xr-x
    * 因此有两种解决方法：一是更改文件权限或所属用户，二是更改本地用户名。
    * 这里用第二种方法：在定义sparksession之前加上代码（放到main函数第一行就行）：System.setProperty("HADOOP_USER_NAME", "root")，第二个参数就是文件所属用户的名字。
    *
    * */
    ssc.checkpoint("hdfs://master:9000/cp-2020-7-15")

    // 提前启动好nc -l -p 8888服务
    val dStream: ReceiverInputDStream[String] = ssc.socketTextStream("master",8888, StorageLevel.MEMORY_AND_DISK)
    val tups: DStream[(String, Int)] = dStream.flatMap(_.split(" ")).map((_,1))
    //调用updateStateByKey原语进行批次累加
    val res: DStream[(String, Int)] = tups.updateStateByKey(func, new HashPartitioner(ssc.sparkContext.defaultParallelism), true)

    res.print()
    ssc.start()
    ssc.awaitTermination()

  }

  //看源码：(iterator: Iterator[(K, Seq[V], Option[S])])其中K为要输入的是键，所以我们定义的时候K设置为自己K的类型String，
  // seq[V]即同批次的key对应的v，其中V也写为自己v对应的类型，OPtions[S]的S也是类型。主要是解释V,S在写代码的时候填为对应的类型
  val func = (it: Iterator[(String, Seq[Int], Option[Int])]) => {
    it.map(x => {
      ( x._1, x._2.sum + x._3.getOrElse(0))//坑：注意这里不能写x._2.reduce(_+_)，否则输入为空会报错
    })
  }
}
